package com.example.future;

import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.RandomUtil;
import com.google.common.collect.ImmutableMap;
import lombok.SneakyThrows;
import lombok.val;
import org.apache.commons.lang3.ThreadUtils;
import org.apache.kafka.common.protocol.types.Field;
import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.ParallelFlux;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.IntUnaryOperator;
import java.util.function.Supplier;
import java.util.stream.IntStream;

/**
 * test
 *
 * @author qingyan 2022-01-19
 */
public class FutureTest {
    Supplier<String> func = () -> {
        randomSleep();
        String string = UUID.randomUUID().toString();
        System.out.println("获取值：" + string);
        return string;
    };

    ThreadPoolExecutor executor = ThreadUtil.newExecutor(10, 20);

    @SneakyThrows
    @Test
    public void test() {
        ForkJoinPool pool = ForkJoinPool.commonPool();
        CompletableFuture<String> future = CompletableFuture.completedFuture(func.get());
        future.whenCompleteAsync((s, throwable) -> log(s));
        log(pool);
        join();
    }
    @Test
    public void test1() {
        CompletableFuture.anyOf(CompletableFuture.supplyAsync(func), CompletableFuture.supplyAsync(func))
                .thenAccept(FutureTest::log);
        join();
    }
    @SneakyThrows
    @Test
    public void test2(){
        IntUnaryOperator func = e -> {
            randomSleep();
            log(e);
            return e * e;
        };
        ForkJoinPool pool = new ForkJoinPool(30);
        ForkJoinTask<Integer> forkJoinTask = pool.submit(() -> {
            return IntStream.range(1, 100).parallel().map(func).reduce(0, (left, right) -> left += right);
        });
        Integer reduce = forkJoinTask.join();
        System.out.println(reduce);
    }


    /**
     * 设置睡眠时间
     */
    @SneakyThrows
    public static void randomSleep() {
        int randomInt = RandomUtil.randomInt(5);
        TimeUnit.SECONDS.sleep(randomInt);
        System.out.println("睡了" + randomInt + "秒");
    }

    @SneakyThrows
    static void join(){
        Thread.currentThread().join();
    }

    public static void log(Object val) {
        System.out.println(Thread.currentThread().getName() + ":" + val.toString());
    }


}
